基於昨天可以把上市公司的股價爬蟲寫出來,
今天把上櫃公司也做一遍~
延續昨天的那份程式,上面這邊套件引用、圖表設定的,影響不大的都保留著
import pandas as pd
import requests
import re
import time
import mplfinance as mpf
import matplotlib.pyplot as plt
import matplotlib.font_manager as fm
from datetime import datetime
# 設定字體
zhfont = fm.FontProperties(fname=r'C:\Windows\Fonts\kaiu.ttf') # 使用原始字串
plt.rcParams['font.family'] = zhfont.get_name()
http://www.tpex.org.tw/web/stock/aftertrading/daily_trading_info/st43_result.php?d=112/08&stkno=6223
json_data=requests.get('http://www.tpex.org.tw/web/stock/aftertrading/daily_trading_info/st43_result.php?d=112/08&stkno=6223').json()
columns= ['日期', '成交股數', '成交金額', '開盤價', '最高價', '最低價', '收盤價', '漲跌價差', '成交筆數']
#只取json格式資料的aaData內的資料哦!
data=pd.DataFrame(json_data['aaData'],columns=columns)
data.head()
日期 成交股數 成交金額 開盤價 最高價 最低價 收盤價 漲跌價差 成交筆數
0 112/08/01 5,583 1,143,509 207.00 213.00 199.00 203.00 -2.00 4,562
1 112/08/02 6,643 1,276,902 203.00 203.50 185.50 187.50 -15.50 5,802
2 112/08/04 3,463 659,699 190.00 194.00 187.00 190.50 3.00 3,020
3 112/08/07 2,752 529,006 190.50 195.50 187.00 194.00 3.50 2,483
4 112/08/08 2,140 409,123 194.00 194.50 189.00 191.00 -3.00 2,326
OK~ 可以正常運作! YES!
(其實主要是懶癌發作(? )
定義一個根據上面下載"上市公司"資料的function
# 定義下載資料的函數
def fetch_data(date, stock_no):
url = f'http://www.twse.com.tw/exchangeReport/STOCK_DAY?response=json&date={date}&stockNo={stock_no}'
max_retries = 5
for i in range(max_retries):
try:
response = requests.get(url)
response.raise_for_status() # 如果請求失敗,則引發異常
return response.json()
except requests.exceptions.RequestException as e:
print(f'Error fetching data for {date}: {e}. Retry {i + 1}/{max_retries}')
time.sleep(3)
raise Exception(f'Failed to fetch data for {date} after {max_retries} retries')
模仿上面的寫法稍微修改一下,
def fetch_tpex_data(date, stock_no):
url = f'https://www.tpex.org.tw/web/stock/aftertrading/daily_trading_info/st43_result.php?d={date}&stkno={stock_no}'
max_retries = 5
for i in range(max_retries):
try:
response = requests.get(url)
response.raise_for_status()
return response.json()
except requests.exceptions.RequestException as e:
print(f'Error fetching TPEX data for {date}: {e}. Retry {i + 1}/{max_retries}')
time.sleep(3)
raise Exception(f'Failed to fetch TPEX data for {date} after {max_retries} retries')
昨天下載資料的function:
# 定義下載和合併資料的函數
def download_stock_data(start_year, start_month, end_year, end_month, stock_no='2330'):
data_frames = []
for year in range(start_year, end_year + 1):
start_m = start_month if year == start_year else 1
end_m = end_month if year == end_year else 12
for month in range(start_m, end_m + 1):
date_str = f'{year}{month:02d}01'
try:
json_data = fetch_data(date_str, stock_no)
columns = ['日期', '成交股數', '成交金額', '開盤價', '最高價', '最低價', '收盤價', '漲跌價差', '成交筆數']
df = pd.DataFrame(json_data['data'], columns=columns)
data_frames.append(df)
except Exception as e:
print(f'Error fetching data for {date_str}: {e}')
# 跳過未來的月份
current_date = datetime.now()
if year == current_date.year and month >= current_date.month:
break
all_data = pd.concat(data_frames, ignore_index=True)
return all_data
那我就在中間抓json資料的地方寫個簡單的if else
判斷一下
是TWSE
就是上市,不是TWSE
就是TPEX
if market == 'TWSE':
json_data = fetch_twse_data(date_str, stock_no)
columns = ['日期', '成交股數', '成交金額', '開盤價', '最高價', '最低價', '收盤價', '漲跌價差', '成交筆數']
df = pd.DataFrame(json_data['data'], columns=columns)
else: # TPEX
json_data = fetch_tpex_data(date_str, stock_no)
columns = ['日期', '成交股數', '成交金額', '開盤價', '最高價', '最低價', '收盤價', '漲跌價差', '成交筆數']
df = pd.DataFrame(json_data['aaData'], columns=columns)
data_frames.append(df)
完整寫起來就長這樣
直接把stock_no
(股票代號)、market
(市場別)的參數設定都放到主程序那,可自由調整
def download_stock_data(start_year, start_month, end_year, end_month, stock_no, market):
data_frames = []
for year in range(start_year, end_year + 1):
start_m = start_month if year == start_year else 1
end_m = end_month if year == end_year else 12
for month in range(start_m, end_m + 1):
date_str = f'{year}{month:02d}01' if market == 'TWSE' else f'{year-1911}/{month:02d}'
try:
if market == 'TWSE':
json_data = fetch_twse_data(date_str, stock_no)
columns = ['日期', '成交股數', '成交金額', '開盤價', '最高價', '最低價', '收盤價', '漲跌價差', '成交筆數']
df = pd.DataFrame(json_data['data'], columns=columns)
else: # TPEX
json_data = fetch_tpex_data(date_str, stock_no)
columns = ['日期', '成交股數', '成交金額', '開盤價', '最高價', '最低價', '收盤價', '漲跌價差', '成交筆數']
df = pd.DataFrame(json_data['aaData'], columns=columns)
data_frames.append(df)
except Exception as e:
print(f'Error fetching data for {date_str}: {e}')
current_date = datetime.now()
if year == current_date.year and month >= current_date.month:
break
all_data = pd.concat(data_frames, ignore_index=True)
return all_data
因為TWSE
和TPEX
提供的格式有點不太一樣,
原本照著寫會出錯,debug了一下,來為上櫃特別寫
if market == 'TPEX':
# 民國年轉換為公元年
def convert_to_gregorian(date_str):
parts = date_str.split('/')
parts[0] = str(int(parts[0]) + 1911)
return '/'.join(parts)
all_data['日期'] = all_data['日期'].apply(convert_to_gregorian)
完整寫起來變成這樣
def clean_data(all_data, market):
if market == 'TPEX':
# 民國年轉換為公元年
def convert_to_gregorian(date_str):
parts = date_str.split('/')
parts[0] = str(int(parts[0]) + 1911)
return '/'.join(parts)
all_data['日期'] = all_data['日期'].apply(convert_to_gregorian)
all_data[['成交股數', '成交金額', '成交筆數']] = all_data[['成交股數', '成交金額', '成交筆數']].replace(',', '', regex=True)
def to_float(x):
try:
return float(x)
except ValueError:
return float('nan')
for col in all_data.columns[1:]:
all_data[col] = all_data[col].map(to_float)
all_data[['成交股數', '成交金額']] = all_data[['成交股數', '成交金額']] / 1000
all_data = all_data.rename(columns={'成交股數': '成交張數'})
all_data = all_data.dropna()
# 轉換日期格式
try:
all_data['日期'] = pd.to_datetime(all_data['日期'], format='%Y/%m/%d')
except ValueError:
all_data['日期'] = pd.to_datetime(all_data['日期'], format='%Y/%m/%d', errors='coerce')
all_data = all_data.dropna(subset=['日期'])
all_data = all_data.rename(columns={
'開盤價': 'Open',
'最高價': 'High',
'最低價': 'Low',
'收盤價': 'Close',
'成交張數': 'Volume'
})
all_data['Open'] = all_data['Open'].astype(float)
all_data['High'] = all_data['High'].astype(float)
all_data['Low'] = all_data['Low'].astype(float)
all_data['Close'] = all_data['Close'].astype(float)
all_data['Volume'] = all_data['Volume'].astype(float)
all_data.set_index('日期', inplace=True)
return all_data
後面主程序的部分就可以自由調整想要的日期區間、股票代號、市場別
# 主程序
start_year = 2019
start_month = 1
end_year = 2024
end_month = 12
stock_no = '6223'
market = 'TPEX'
#呼叫函式
all_data = download_stock_data(start_year, start_month, end_year, end_month, stock_no, market)
cleaned_data = clean_data(all_data, market)
繪圖記得給他畫出來
import mplfinance as mpf
# 繪製 K 線圖
def plot_data(all_data, title):
fig, ax = plt.subplots()
# 自定義樣式
custom_style = mpf.make_mpf_style(
base_mpf_style='charles',
marketcolors={
'candle': {'up': 'red', 'down': 'green'},
'edge': {'up': 'red', 'down': 'green'},
'wick': {'up': 'red', 'down': 'green'},
'ohlc': {'up': 'red', 'down': 'green'},
'volume': {'up': 'red', 'down': 'green'},
'alpha': 1.0
}
)
# 繪圖
mpf.plot(all_data, type='candle', style=custom_style, ax=ax, warn_too_much_data=len(all_data) + 1)
ax.set_title(title, fontproperties=zhfont)
ax.set_ylabel('價格', fontproperties=zhfont)
plt.show()
最後plot
出來
plot_data(cleaned_data, f'旺矽 {start_year}/{start_month} - {end_year}/{end_month} 每日股票交易價格和收盤情況')
每日記錄:
明天又要開盤了,看戲 (吃爆米花
其實K線圖畫出來,如果自己想要進行回測或是寫策略就可以在後面加上去~
或是其實用其他人家軟體、平台提供寫好的工具進行回測和寫策略也可以,
個人習慣問題,自己寫比較苦,但是很自由。
然後變成程式交易(誤
之後要是後面做完還沒滿三十天,來放一些debug的過程好了(陷入重重危機)。
恭喜林郁婷為她也為台灣奪下史上女子拳擊第一面奧運金牌!!
當下看到那一刻只有滿滿的感動,瞬間熱淚盈眶,
小時候我也被霸凌過,要怎麼想像一個女生前面27年來是怎麼承受這些耳語的,
她代表著台灣,用善良,來對抗世界的霸凌。 🥹